package com.jianying.aws;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class mysqlSource {
    public static void main(String[] args) throws Exception {
        // 创建一个配置对象
        Configuration conf = new Configuration();
        // 启用 Flink Web UI
        conf.set(RestOptions.PORT,8088);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("database-mysql-1.cfgi4aw8kw3z.rds.cn-north-1.amazonaws.com.cn")
                .port(3306)
                .databaseList("mytest") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("mytest.user_events") // set captured table
                .username("root")
                .password(".Ljy12qw")
                .serverTimeZone("Asia/Shanghai")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();

        // enable checkpoint
        env.enableCheckpointing(3000);

        env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
                // set 4 parallel source tasks
                .setParallelism(1)
                .print().setParallelism(1); // use parallelism 1 for sink to keep message ordering

        env.execute("Print MySQL Snapshot + Binlog");

    }
}
